// Copyright (c) 2020 Presto Labs Pte. Ltd.
// Author: donggu

#include "coin2/exchange/okex_common/feed_v3/processor.h"

#include <string_view>
#include <utility>

#include "coin2/base/string_util.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/okex_common/feed_v3/update.h"
#include "coin2/exchange/util/json_util.h"

using namespace std::literals;

namespace coin2::exchange::okex_common::feed_v3::impl {

using coin2::exchange::okex_common::feed_v3::OkexCommonFeedUpdateParser;

bool OkexCommonFeedProcessor::CheckChannelFilter(const std::string& ch) {
  // index is rest api, its format is different with websocket api
  if (FeedParsingProcessor::CheckChannelFilter(std::string(ch))) {
    return true;
  } else {
    return FeedParsingProcessor::CheckChannelFilter(std::string(ch) + ":{}");
  }
}

// we share the same book builder between depth5, depth400 snapshot, depth400 diff
void OkexCommonFeedProcessor::OnTopicRecord(const base::executor::TopicRecordPacket& packet) {
  if (packet.payload == "pong") return;
  upd_mgr_.onBeginPacket(packet);

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(packet.payload);
  }

  simdjson::dom::object obj;
  simdjson::error_code err;
  parser_.parse(packet.payload).get<simdjson::dom::object>().tie(obj, err);

  if (err) {
    Error("[Okex Common Feed Processor] err", packet.payload);
    return;
  }

  std::string_view table;
  obj["table"].get<std::string_view>().tie(table, err);
  if (err) {
    //  {"event":"subscribe","channel":"spot/depth_l2_tbt:BAT-USDT"}

    // guess channel by fields
    if (obj["index"].error() != simdjson::NO_SUCH_FIELD) {
      table = "index"sv;
    } else if (obj["url"].error() != simdjson::NO_SUCH_FIELD) {
      // guess candles channel by url
      std::string_view url = obj["url"].get<std::string_view>();
      size_t sep = url.rfind("/");
      size_t first_and = url.find("&", sep);
      table = url.substr(sep, first_and - sep -1);
    } else {
      return;
    }
  }

  // parse this msg?
  if (!CheckChannelFilter(std::string(table))) {
    return;
  }

  simdjson::dom::array data;
  obj["data"].get<simdjson::dom::array>().tie(data, err);
  if (err && (table != "index")) {
    Error("[Okex Common Feed Processor] extract data error", packet.payload);
    return;
  }

  if (StringEndsWith(table, "/trade"sv)) {
    std::string_view last_symbol;
    Trade* agg_trade = upd_mgr_.GetTradeMutable();
    agg_trade->Clear();

    // trades:
    //   symbol, ts, p, q, side
    //   symbol, ts, p, q, side
    //   symbol, ts, p, q, side
    // ...

    Trade trade{};
    for (const auto& item : data) {
      std::string_view native_symbol = item["instrument_id"].get<std::string_view>();
      if (!upd_mgr_.is_symbol_registered(native_symbol)) {
        continue;
      }
      OkexCommonFeedUpdateParser::ParseTrade(item, &trade);
      if (native_symbol == last_symbol && TryAggregateTrades(agg_trade, trade)) {
        continue;
      }
      if (agg_trade->fill_qty > 0) {
        CHECK_GT(last_symbol.size(), 0);
        upd_mgr_.PublishTrade(last_symbol, true);
      }
      *agg_trade = trade;
      last_symbol = native_symbol;
    }
    if (agg_trade->fill_qty > 0) {
      upd_mgr_.PublishTrade(last_symbol, false);
    }
  } else if (StringEndsWith(table, "/depth5"sv)) {
    for (const auto& item : data) {
      std::string_view native_symbol = item["instrument_id"].get<std::string_view>();
      auto book = upd_mgr_.GetBookMutable(native_symbol, "depth5");
      if (book != nullptr) {
        if (publish_bbo_book_only_) {
          std::optional<PriceQty> bid_pre = book->Bid0();
          std::optional<PriceQty> ask_pre = book->Ask0();
          OkexCommonFeedUpdateParser::ParseBook(book, item, BookBuilder::LIGHT_BBO);
          std::optional<PriceQty> bid_post = book->Bid0();
          std::optional<PriceQty> ask_post = book->Ask0();
          bool is_bbo_changed = (bid_pre != bid_post) || (ask_pre != ask_post);
          if (is_bbo_changed) {
            upd_mgr_.PublishBook();
          }
        } else {
          OkexCommonFeedUpdateParser::ParseBook(book, item, BookBuilder::DELTA);
          upd_mgr_.PublishBook();
        }
      }
    }
  } else if (
      // depth 400(futures) or depth 200(swap)
      StringEndsWith(table, "/depth"sv) || StringEndsWith(table, "/depth_l2_tbt"sv)) {
    // insert ?
    // update (inst id, asks, bids, timestamp, checksum) // compatible with depth5
    // delete ?
    // partial ()

    std::string_view action = obj["action"];
    if (action == "update") {
      for (const auto item : data) {
        std::string_view native_symbol = item["instrument_id"].get<std::string_view>();
        auto book = upd_mgr_.GetBookMutable(native_symbol, "depth");
        if (book != nullptr) {
          if (parser_config_.has_sampling_frequency_bps()) {
            OkexCommonFeedUpdateParser::ParseBook(book, item, BookBuilder::LIGHT_BBO);
            if (book->SampleByFrequency(
                    book->Bid0(),
                    book->Ask0(),
                    parser_config_.sampling_frequency_bps())) {
              upd_mgr_.PublishBook();
            }
          } else if (publish_bbo_book_only_) {
            std::optional<PriceQty> bid_pre = book->Bid0();
            std::optional<PriceQty> ask_pre = book->Ask0();
            OkexCommonFeedUpdateParser::ParseBook(book, item, BookBuilder::LIGHT_BBO);
            std::optional<PriceQty> bid_post = book->Bid0();
            std::optional<PriceQty> ask_post = book->Ask0();
            bool is_bbo_changed = (bid_pre != bid_post) || (ask_pre != ask_post);
            if (is_bbo_changed) {
              upd_mgr_.PublishBook();
            }
          } else {
            OkexCommonFeedUpdateParser::ParseBook(book, item, BookBuilder::DELTA);
            upd_mgr_.PublishBook();
          }
        }
      }
    } else if (action == "partial") {
      for (const auto item : data) {
        std::string_view native_symbol = item["instrument_id"].get<std::string_view>();
        auto book = upd_mgr_.GetBookMutable(native_symbol, "depth");
        if (book != nullptr) {
          if (parser_config_.has_sampling_frequency_bps()) {
            OkexCommonFeedUpdateParser::ParseBook(book, item, BookBuilder::LIGHT_BBO);
            if (book->SampleByFrequency(
                    book->Bid0(),
                    book->Ask0(),
                    parser_config_.sampling_frequency_bps())) {
              upd_mgr_.PublishBook();
            }
          } else if (publish_bbo_book_only_) {
            std::optional<PriceQty> bid_pre = book->Bid0();
            std::optional<PriceQty> ask_pre = book->Ask0();
            OkexCommonFeedUpdateParser::ParseBook(book, item, BookBuilder::LIGHT_BBO);
            std::optional<PriceQty> bid_post = book->Bid0();
            std::optional<PriceQty> ask_post = book->Ask0();
            bool is_bbo_changed = (bid_pre != bid_post) || (ask_pre != ask_post);
            if (is_bbo_changed) {
              upd_mgr_.PublishBook();
            }
          } else {
            OkexCommonFeedUpdateParser::ParseBook(book, item, BookBuilder::SNAPSHOT);
            upd_mgr_.PublishBook();
          }
        }
      }
    } else {
      Error("[Okex Common Processor] not implmeneted for " + std::string(table), packet.payload);
    }
  } else if (StringEndsWith(table, "/ticker"sv)) {
    for (const auto item : data) {
      std::string_view native_symbol = item["instrument_id"].get<std::string_view>();
      if (upd_mgr_.is_symbol_registered(native_symbol)) {
        auto* open_interest = upd_mgr_.GetOpenInterestMutable();
        if (OkexCommonFeedUpdateParser::ParseOpenInterest(item, open_interest)) {
          upd_mgr_.PublishOpenInterest(native_symbol);
        }
      }
    }
  } else if (StringEndsWith(table, "/mark_price"sv)) {
    for (const auto item : data) {
      std::string_view native_symbol = item["instrument_id"].get<std::string_view>();
      if (upd_mgr_.is_symbol_registered(native_symbol)) {
        auto* mark_price = upd_mgr_.GetMarkPriceMutable();
        OkexCommonFeedUpdateParser::ParseMarkPrice(item, mark_price);
        upd_mgr_.PublishMarkPrice(native_symbol);
      }
    }
  } else if (StringEndsWith(table, "/funding_rate"sv)) {
    for (const auto item : data) {
      std::string_view native_symbol = item["instrument_id"].get<std::string_view>();
      if (upd_mgr_.is_symbol_registered(native_symbol)) {
        auto* funding_rate = upd_mgr_.GetFundingRateMutable();
        OkexCommonFeedUpdateParser::ParseFundingRate(item, funding_rate, packet.timestamp);
        upd_mgr_.PublishFundingRate(native_symbol);
      }
    }
  } else if (table == "index") {
    std::string_view native_symbol = obj["instrument_id"].get<std::string_view>();
    if (upd_mgr_.is_symbol_registered(native_symbol)) {
      auto* index = upd_mgr_.GetIndexMutable();
      index->timestamp = TimestampFromOkexIndexIso8601(obj["timestamp"].get<std::string_view>());
      index->price = stod(obj["index"].get<const char*>());
      upd_mgr_.PublishIndex(native_symbol);
    }
  } else if (StringStartsWith(table, "/candles"sv)) {
    // rest kline only support 1 worker, worker string is behind group
    if (!long_topic_.has_value()) {
      if (packet.topic_name == upd_mgr_.mea().SnakeString()) {
        long_topic_ = false;
      } else {
        long_topic_ = true;
      }
    }
    // only long topic contains worker setting
    if (long_topic_.value()) {
      int pos = packet.topic_name.find_last_of('_');
      int worker = std::stoi(packet.topic_name.substr(pos + 2));
      if (picked_worker) {
        if (worker != picked_worker) return;
      } else {
        picked_worker = worker;
      }
    }
    std::string_view url = obj["url"].get<std::string_view>();
    std::string native_symbol, granularity;
    int64_t start_timestamp;
    int page;
    std::tie(native_symbol, granularity, start_timestamp, page) =
        OkexCommonFeedUpdateParser::ExplodeUrl(url);
    fastfeed::proto::KlineInterval interval =
        OkexCommonFeedUpdateParser::ParseKlineInterval(granularity);
    auto* kline = upd_mgr_.GetKlineMutable(native_symbol, interval);
    if (kline != nullptr) {
      // rest api, kline is descending
      bool to_publish = false;
      // query is split up several requests
      if (page == 0) {
        // first long query
        kline->SnapshotClear();
      } else {
        kline->UpdateClear();
      }
      for (int i = data.size() -1; i >= 0; i--) {
        simdjson::dom::array kline_data = data.at(i);
        bool freezed = OkexCommonFeedUpdateParser::Parse1Kline(kline_data, kline,
            packet.timestamp, interval);
        to_publish = to_publish || freezed;
      }

      if (to_publish) {
        upd_mgr_.PublishKline(native_symbol);
      }
    }
  } else if (StringStartsWith(table, "swap/candle"sv) ||
             StringStartsWith(table, "futures/candle"sv)) {
    for (const auto& item : data) {
      std::string_view native_symbol = item["instrument_id"].get<std::string_view>();
      // table : swap/candle60s, futures/candle60s
      std::vector<std::string> splited = Split(table, "/");
      CHECK_EQ(splited.size(), 2);
      std::string& candle = splited.at(1);
      std::string_view granularity = candle.substr(6, candle.size() - 6 -1);
      fastfeed::proto::KlineInterval interval =
          OkexCommonFeedUpdateParser::ParseKlineInterval(granularity);
      auto* kline = upd_mgr_.GetKlineMutable(native_symbol, interval);
      if (kline != nullptr) {
        simdjson::dom::array kline_data = item["candle"];
        kline->UpdateClear();
        if (OkexCommonFeedUpdateParser::Parse1Kline(kline_data, kline, packet.timestamp,
                interval)) {
          upd_mgr_.PublishKline(native_symbol);
        }
      }
    }
  }

  upd_mgr_.onEndPacket(packet);
}
}  // namespace coin2::exchange::okex_common::feed_v3::impl
